package mspbots.hubspot.client.core;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;
import mspbots.hubspot.client.oauth.Authorization;
import mspbots.hubspot.common.RabbitKeys;
import mspbots.hubspot.common.RedisKeys;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitMessagingTemplate;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Optional;

/**
 * data.connector.hubspot.api.AsyncScheduleEvent
 *
 * @author Alex bob(https://github.com/vnobo)
 * @date Created by 2020/4/3
 */
@Log4j2
@Component
@RequiredArgsConstructor
public class GlobalScheduledTimer {

    private final Map<String, BaseSyncService> syncServiceMap;
    private final ReactiveStringRedisTemplate stringRedisTemplate;
    private final ReactiveRedisTemplate<String, Object> reactiveRedisTemplate;
    private final RabbitMessagingTemplate rabbitMessagingTemplate;
    private final ObjectMapper objectMapper;

    private Flux<Authorization> getCache() {
        return this.reactiveRedisTemplate.scan(ScanOptions.scanOptions()
                .match(RedisKeys.AUTHORIZE_REDIS_KEY + "*")
                .count(100).build())
                .flatMap(key -> this.reactiveRedisTemplate.opsForValue().get(key))
                .map(data -> this.objectMapper.convertValue(data, Authorization.class));
    }

    @Scheduled(fixedDelay = 6000 * 60 * 8)
    public void autoSyncHour() {
        Optional<List<Authorization>> authorizationListOpt = this.getCache().collectList().blockOptional(Duration.ofSeconds(10));
        authorizationListOpt.ifPresent(authorizations -> authorizations.parallelStream().forEach(property -> {
            try {
                this.syncServiceMap.values().parallelStream()
                        .forEach(v -> v.syncProperties(property.getTenantId()).block());
            } catch (Exception e) {
                log.error("Sync {} property task error, msg: {}", property.getTenantId(), e.getMessage());
            }
        }));
    }

    @Scheduled(fixedDelay = 1000 * 60 * 5)
    public void autoSync() {
        Boolean lock = this.stringRedisTemplate.opsForValue().get(RedisKeys.TASK_LOCK_ANCHOR_REDIS_KEY)
                .map(Boolean::parseBoolean).block(Duration.ofSeconds(10));
        if (ObjectUtils.isEmpty(lock)) {
            this.stringRedisTemplate.opsForValue()
                    .setIfAbsent(RedisKeys.TASK_LOCK_ANCHOR_REDIS_KEY, "false", Duration.ofMinutes(5))
                    .block(Duration.ofSeconds(10));

            this.getCache().filterWhen(property -> this.stringRedisTemplate.opsForValue()
                    .size(RedisKeys.ON_OFF_TENANT_TASK_REDIS_KEY + ":" + property.getTenantId())
                    .defaultIfEmpty(0L).map(count -> count == 0))
                    .subscribe(property -> {
                        log.debug("Auto sync send properties {}", property);
                        this.rabbitMessagingTemplate.convertAndSend(RabbitKeys.BASE_EXCHANGE_NAME,
                                RabbitKeys.DISTRIBUTED_TASK_QUEUE_KEY, property);
                    });
        }
    }

    @RabbitListener(queues = "hub-spot.distributed.task.sync", concurrency = "10")
    public void syncTask(Authorization properties) {

        String tenantSyncTaskLockKey = RedisKeys.RUNNING_TASK_LOCK_ANCHOR_REDIS_KEY + ":" + properties.getTenantId();

        Optional<Long> countOpt = this.stringRedisTemplate.opsForValue()
                .size(tenantSyncTaskLockKey).blockOptional(Duration.ofSeconds(10));

        if (countOpt.isPresent() && countOpt.get() > 0) {
            log.info("*********Sync {} Task running*************", properties.getTenantId());
            return;
        }

        log.info("===Sync {} Task start===", properties.getTenantId());
        this.reactiveRedisTemplate.opsForValue()
                .setIfAbsent(tenantSyncTaskLockKey, properties.getTenantId(), Duration.ofMinutes(5))
                .block(Duration.ofSeconds(10));

        try {

            this.syncServiceMap.values().parallelStream()
                    .forEach(v -> v.syncData(properties.getTenantId()).block());

        } catch (Exception e) {
            log.error("Sync {} task error, msg: {}", properties.getTenantId(), e.getMessage());
        }

        this.reactiveRedisTemplate.opsForValue().delete(tenantSyncTaskLockKey).block(Duration.ofSeconds(10));
        log.info("===Sync {} Task end===", properties.getTenantId());
    }

}
